use crossbeam_deque::Injector;
use fudata::db;
use fudata::model::kline::KlineList;
use log::error;
use log::trace;
use std::sync::Arc;
use anyhow::anyhow;
use anyhow::Result;
use fudata::model::security::SecurityList;
use log::info;
use zenoh::key_expr::KeyExpr;
use zenoh::prelude::r#async::*;
use zenoh::queryable::Query;

use super::RECEIVE_TIMEOUT_MS;

// Create a worker pool using Tokio
pub fn start_worker_pool(num_workers: u8) -> Arc<Injector<Query>> {
    let injector = Arc::new(Injector::new());

    // Spawn workers directly using tokio::spawn
    for worker_no in 0..num_workers {
        let ij_clone = Arc::clone(&injector);
        tokio::spawn(async move {
            loop {
                let task = ij_clone.steal().success();

                match task {
                    Some(todo) => {
                        // Process the todo using the handle_query function
                        let _sample = handle_query(worker_no, &todo).await;
                        // Do something with the sample...
                    }
                    None => {
                        trace!("(worker_query) Workder[{worker_no}] steal no todo, wait {}ms", RECEIVE_TIMEOUT_MS);
                        tokio::time::sleep(
                            std::time::Duration::from_millis(RECEIVE_TIMEOUT_MS)
                        ).await;
                    }
                }
            }
        });
    }

    injector
}

// Define the handle_query function
async fn handle_query(worker_no: u8, query: &Query) -> Result<()> {
    // Your actual query processing logic goes here
    // ...

    let selector = query.selector();
    info!(
        "\n\n (handle_query) Worker[{worker_no}] Received query with selector key_expr={:?} parameters={} selector.parameters_stringmap()={:?}\n\n",
        selector.key_expr,
        selector.parameters(),
        selector.parameters_stringmap()
    );
    let key = selector.key_expr.as_str();
    match key {
        "securities" => {
            handle_query_securites(&query).await?;
        }
        "klines" => {
            handle_query_klines(&query).await?;
        }
        _ => {
            error!("unknown key_expr:{}", key);
        }
    }

    Ok(())
}

async fn handle_query_securites(query: &Query) -> Result<()> {
    let selector = query.selector();

    let mut last_updated: Option<String> = None;
    // Extract parameters from the selector
    match selector.parameters_stringmap() {
        Ok(params) => {
            if let Some(value) = params.get("_time") {
                // Note: Using "_time" as placeholder; correct key should be used
                info!("Found parameter '_time': {}", value);
            }
            if let Some(ts) = params.get("last_updated") {
                info!("Last updated parameter: {}", ts);
                last_updated = Some(ts.to_string());
            }
        }
        Err(err) => {
            error!("Failed to parse parameters: {}", err);
        }
    }
    let securities = db::security::select(&None, None, last_updated, None, 0, 10).await?;
    let ss: zenoh::value::Value = SecurityList::from(securities).into();
    let sample = Sample::new(KeyExpr::try_from("securities").unwrap(), ss);
    let rs = query.reply(Ok(sample)).res().await;

    match rs {
        Ok(_) => Ok(()),
        Err(e) => Err(anyhow!(e)),
    }
}

async fn handle_query_klines(query: &Query) -> Result<()> {
    let selector = query.selector();
    let mut security_id: Option<String> = None;
    let mut last_updated: Option<String> = None;
    // Extract parameters from the selector
    match selector.parameters_stringmap() {
        Ok(params) => {
            if let Some(value) = params.get("security_id") {
                // Note: Using "_time" as placeholder; correct key should be used
                info!("Found parameter 'security_id': {}", value);
                security_id = Some(value.to_string());
            }
            if let Some(ts) = params.get("last_updated") {
                info!("Last updated parameter: {}", ts);
                last_updated = Some(ts.to_string());
            }
        }
        Err(err) => {
            error!("Failed to parse parameters: {}", err);
        }
    }

    let ss = match security_id {
        Some(secc_id) => {
            let klines = db::kline::select(
                &secc_id,
                last_updated,
                None,
                Some("last_updated ASC"),
                0,
                10
            ).await?;
            // info!("{:?}", klines);
            let ss: zenoh::value::Value = KlineList::from(klines).into();
            ss
        }
        None => {
            error!("(handle_query_klines) No parameter: security_id");

            let ss: zenoh::value::Value = KlineList::from(vec![]).into();
            ss
        }
    };

    let sample = Sample::new(KeyExpr::try_from("klines").unwrap(), ss);
    let rs = query.reply(Ok(sample)).res().await;

    match rs {
        Ok(_) => Ok(()),
        Err(e) => Err(anyhow!(e)),
    }
}
